package com.jiayuan.cn.energy.test.ok;

import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.streaming.api.collector.selector.OutputSelector;
//import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
//import org.apache.flink.streaming.connectors.redis.RedisSink;
//import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
//import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
//import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
//import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
//import org.apache.kafka.clients.consumer.ConsumerConfig;
//import org.apache.kafka.clients.producer.ProducerConfig;
//import org.apache.kafka.common.serialization.StringDeserializer;
//import org.apache.kafka.common.serialization.StringSerializer;


public class FlinkToMysql {
    private static String MYSQL_SERVER = "192.168.110.88";

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment
                .getExecutionEnvironment();

//        DataStream<String> mySource = environment.readTextFile("/path/dataval.txt");
//        DataStream<String> mySource = environment.readTextFile("/path/dataval.csv");
//        DataStreamSource<DataV> mySource = environment.fromData(new DataV(1, "s1", 11.1), new DataV(2, "s2", 22.2));
//        DataV [] dataVs = new DataV[2];
//        dataVs[0] = new DataV(1, "s1", 11.1);
//        dataVs[1] = new DataV(2, "s2", 22.2);
//        DataStreamSource<DataV> mySource = environment.fromData(dataVs);
        DataStreamSource<String> mySource = environment.fromData("Hello flink", "Hello world","Hello elves","Hello cn");
//        String sql = "insert into dataval(id,username,price) values  " + "(?,?,?)";
        String sql = "insert into test1(str) values  " + "(?)";
        JdbcConnectionOptions jdbcBuild = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
//                .withDriverName("com.mysql.jdbc.Driver")
                .withDriverName("com.mysql.cj.jdbc.Driver")
                .withUrl("jdbc:mysql://192.168.110.88:3306/helm?useUnicode=true&characterEncoding=utf-8&useSSL=false")
                .withUsername("helm")
                .withPassword("db-999")
                .build();
//        new DataV(1, "s1", 11.1).getId();
        mySource.addSink(JdbcSink.sink(sql, (ps, dataV) -> {
//            ps.setInt(1, dataV.getId());
//            ps.setString(2, dataV.getUsername());
//            ps.setDouble(3, dataV.getPrice());
            ps.setString(1, dataV);
        }, jdbcBuild));

        environment.execute("mysql dataval job");

    }


}






